{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "4d31ccac-1fcb-430f-be49-bbe4dde9df7f",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# RDD编程入门"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "1ab228d8-f221-4823-bba5-334b329f6f23",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "环境初始化\n",
    "> 首先执行环境的初始化。  \n",
    "> 将存储账户与Spark环境关联，以便于在Spark程序中可以使用存储。  \n",
    "> `dfs_endpoint` 是文件系统的根端点。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "8a5d5b18-e3f7-4edb-8e55-866f750d6adf",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%run \"../../../initialization\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "2bb245b8-5c68-4d18-a0e2-b56c5888b2ad",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## SparkContext对象\n",
    "\n",
    "Spark RDD 编程的程序入口对象是SparkContext对象(不论何种编程语言)。\n",
    "\n",
    "只有构建出SparkContext，基于它才能执行后续的API调用和计算。\n",
    "\n",
    "本质上，SparkContext对编程来说，主要功能就是创建第一个RDD出来。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "74742dd3-3040-429e-9488-794c8834e7f6",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "23d8b643-8aab-45ec-8ef1-2b93deb46488",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## RDD的创建\n",
    "\n",
    "可以通过以下几种方式创建RDD：\n",
    "* 并行化创建\n",
    "* 读取文件创建"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "1f1d6414-5680-48a6-8ac7-f65a73f8cb77",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### 并行化创建\n",
    "\n",
    "并行化创建，是指将本地集合转换为分布式RDD。\n",
    "\n",
    "并行化创建RDD，使用`parallelize`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "efba0dc4-07dd-4fe6-8ad2-9c85be45fc52",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "list = [1,2,3,4,5,6,7,8,9]\n",
    "print(type(list))\n",
    "print(\"本地集合：\", list)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "22266cf2-b69e-4f4a-8d4b-523538badd5c",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "以下代码通过将本地集合并行化，转换成分布式RDD。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "36a7e288-c3f6-4be1-8b24-2d71efbead60",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize(list, 4)\n",
    "print(type(rdd))\n",
    "print(\"分布式RDD：\", rdd)\n",
    "print(\"分布式RDD：\", rdd.collect())\n",
    "print(\"RDD的分区数：\", rdd.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "bb697b7c-9a02-42e7-ae65-3077dcdded19",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### 读取文件创建\n",
    "\n",
    "读取文件创建，可以使用以下几个API  \n",
    "* textFile\n",
    "* wholeTextFiles"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "bda42989-683e-4a77-9065-abe06a44baaa",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### textFile\n",
    "\n",
    "`textFile`可以读取本地文件，也可以读取分布式文件系统上的文件，比如：`HDFS`、`S3`。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "7a148b78-bcc1-4ddd-9d6c-9e0bed2ac462",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# 读取分布式文件系统上的文件\n",
    "rdd = sc.textFile(\"%s/word.txt\" % dfs_endpoint)\n",
    "\n",
    "print(type(rdd))\n",
    "print(rdd)\n",
    "print(rdd.collect())\n",
    "print(rdd.getNumPartitions())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "e231e1a1-4829-4630-8f57-6574e125f5ef",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# 读取Databricks文件系统(DBFS)上的文件\n",
    "rdd = sc.textFile(\"dbfs:/mnt/databrickscontainer1/word.txt\")\n",
    "\n",
    "print(type(rdd))\n",
    "print(rdd)\n",
    "print(rdd.collect())\n",
    "print(rdd.getNumPartitions())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "8db796c3-5775-4ce4-8a8a-135a73bb12d3",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# 读取Databricks文件系统(DBFS)上的文件\n",
    "# 可以像访问本地文件一样直接使用绝对路径进行访问\n",
    "rdd = sc.textFile(\"/mnt/databrickscontainer1/word.txt\")\n",
    "\n",
    "print(type(rdd))\n",
    "print(rdd)\n",
    "print(rdd.collect())\n",
    "print(rdd.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "d69f542f-b4c8-44a4-9f43-2ba1a6372678",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### wholeTextFiles\n",
    "\n",
    "与`textFile`功能一致，不过`wholeTextFiles`更适合读取很多小文件的场景。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "5b5b8181-fbf0-4091-8e36-9cbe0a01676b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.wholeTextFiles(\"/mnt/databrickscontainer1\")\n",
    "\n",
    "print(type(rdd))\n",
    "print(rdd)\n",
    "print(rdd.collect())\n",
    "print(rdd.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "ea26366a-19fc-49f4-b073-e43aee6d6534",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## RDD算子"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "b6218858-1277-4e9a-8b3c-542c7f91014f",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### 什么是算子\n",
    "\n",
    "分布式集合对象上的API称之为**算子**。\n",
    "\n",
    "本地对象上的API叫做**函数**/**方法**。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "96c55164-f052-497d-b0a7-2959b97d5560",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### 算子的分类\n",
    "\n",
    "RDD的算子分成两类：\n",
    "* Transformation：转换算子\n",
    "* Action：动作算子\n",
    "\n",
    "**Transformation算子：**\n",
    "\n",
    "定义：**返回值是一个RDD**的算子，称为Transformation算子。\n",
    "\n",
    "特性：这类算子是懒加载（lazy）的，如果没有Action算子，Transformation算子是不工作的。\n",
    "\n",
    "**Action算子：**\n",
    "\n",
    "定义：**返回值不是RDD**的算子，称为Action算子。\n",
    "\n",
    "> 对于这两类算子来说，Transformation算子相当于在构建执行计划，Action算子是一个指令让这个执行计划开始工作。  \n",
    "如果没有Action，Transformation算子之间的迭代关系仅仅是一个执行计划，不会执行，只有Action到来，才会按照执行计划开始工作。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "bb8dd5b4-9beb-48cc-86cb-39e3221b5ebc",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### 常用的Transformation算子"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "9578ba21-d500-43ad-a58f-f07c2ad84084",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### map 算子\n",
    "\n",
    "map算子，是将RDD的数据一条一条处理，处理的逻辑是基于map算子中接收的处理函数的，返回新的RDD。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.map(func)\n",
    "```\n",
    "> func: f:(T) -> U  \n",
    "f: 表示这是一个函数/方法  \n",
    "(T) -> U 表示的是方法定义  \n",
    "(T) 表示参数列表  \n",
    "U 表示返回值"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "cbebbc9c-c7f0-485c-bd9a-279c12398bd7",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([0,1,2,3,4,5,6,7,8,9])\n",
    "\n",
    "# 定义func\n",
    "def map_func(data):\n",
    "    return data * 5\n",
    "\n",
    "# rdd.map(func)\n",
    "print(rdd.map(map_func).collect())\n",
    "\n",
    "# 可以直接用匿名函数的方式，lambda表达式\n",
    "print(rdd.map(lambda x: x + 5).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "a7199721-2ec6-4976-b5ee-61a619221957",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### flatMap 算子\n",
    "\n",
    "flatMap算子，是对RDD执行map操作，然后进行解除嵌套操作，返回新的RDD。\n",
    "\n",
    "> **解除嵌套：**  \n",
    "嵌套的数据：  \n",
    "lst = [[1,2,3],[4,5,6],[7,8,9]]  \n",
    "解除嵌套的数据：  \n",
    "lst = [1,2,3,4,5,6,7,8,9]\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.flatMap(func)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "3e2e57b7-2087-4dcf-b84f-dc1ac6202ed0",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([\"1 2 3\",\"4 5 6\",\"7 8 9\"])\n",
    "# rdd有3个元素，map操作后还是3个嵌套元素\n",
    "print(rdd.map(lambda x: x.split(\" \")).collect())\n",
    "# rdd有3个元素，flatMap操作后是无嵌套元素\n",
    "print(rdd.flatMap(lambda x: x.split(\" \")).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "82293a72-7ed5-4912-aa9e-982e20bb46eb",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### mapValues 算子\n",
    "\n",
    "mapValues算子，**针对K-V型RDD**，对二元组的Value执行map操作。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.mapValues(func)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "3b3f27a8-1de4-41df-af60-a0e27e39ed00",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\", 1), (\"b\", 1), (\"a\", 2), (\"b\", 2), (\"a\", 3)])\n",
    "print(rdd.collect())\n",
    "print(rdd.mapValues(lambda x: x * 2).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "6d955108-18f1-41bd-80b4-562be53804b1",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### reduceByKey 算子\n",
    "\n",
    "reduceByKey算子，**针对K-V型RDD**，自动按照K分组，然后根据提供的聚合逻辑，完成组内数据的聚合操作。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.reduceByKey(func)\n",
    "```\n",
    "\n",
    "> func: (V, V) -> V  \n",
    "func 函数只负责处理聚合逻辑，不负责分组"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "f7772a0b-d813-4134-84af-fc3ce44af77e",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\", 1), (\"b\", 1), (\"a\", 2), (\"b\", 2), (\"a\", 3)])\n",
    "print(rdd.reduceByKey(lambda a,b: a + b).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "671f2155-d172-4475-a614-0840a8a79e4e",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### WordCount案例回顾\n",
    "\n",
    "在我们的WordCount案例中，使用到以下知识点：\n",
    "* 程序入口对象是SparkContext对象，sc，主要功能就是创建第一个RDD出来\n",
    "  * wordsRDD = sc.textFile(\"/mnt/databrickscontainer1/word.txt\")\n",
    "* 通过读取文件创建RDD，textFile\n",
    "  * wordsRDD = sc.textFile(\"/mnt/databrickscontainer1/word.txt\")\n",
    "* RDD的特性\n",
    "  * wordsRDD -> flatMapRDD -> mapRDD -> resultRDD\n",
    "* 三个Transformation算子\n",
    "  * flatMap\n",
    "  * map\n",
    "  * reduceByKey"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "dd8db57c-bafc-4c6f-adb3-3ecbdf3e69c5",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# 第一步、读取本地数据 封装到RDD集合，认为列表List\n",
    "wordsRDD = sc.textFile(\"/mnt/databrickscontainer1/word.txt\")\n",
    "# 第二步、处理数据 调用RDD中函数，认为调用列表中的函数\n",
    "# a. 每行数据分割为单词\n",
    "flatMapRDD = wordsRDD.flatMap(lambda line: line.split(\" \"))\n",
    "# b. 转换为二元组，表示每个单词出现一次\n",
    "mapRDD = flatMapRDD.map(lambda x: (x, 1))\n",
    "# c. 按照Key分组聚合\n",
    "resultRDD = mapRDD.reduceByKey(lambda a, b: a + b)\n",
    "# 第三步、输出数据\n",
    "print(resultRDD.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "34ef09a0-2e56-497a-a70a-1465defd1ea8",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### glom 算子\n",
    "\n",
    "glom算子，将RDD的数据，按照数据的分区添加嵌套。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.glom()\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "b6422197-6487-46a5-b9dd-d1dbee53c3b9",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([0,1,2,3,4,5,6,7,8,9])\n",
    "\n",
    "print(rdd.collect())\n",
    "print(rdd.glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "7be63001-15dd-4eac-8d04-70ce2f4018c3",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### groupBy 算子\n",
    "\n",
    "groupBy算子，将RDD的数据进行分组。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.groupBy(func)\n",
    "\n",
    "# func 函数\n",
    "# 函数要求传入一个参数，返回一个返回值\n",
    "# groupBy根据函数的返回值，将具有相同返回值的元素放入同一个组中\n",
    "# 分组完成后的RDD的每个元素都是一个二元组，key是返回值，value是具有相同返回值的原RDD的元素\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "4adce20d-a797-40c0-9867-a985cf84ad0a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([0,1,2,3,4,5,6,7,8,9])\n",
    "# ResultIterable\n",
    "print(rdd.groupBy(lambda x: \"group {}\".format(x % 3)).collect())\n",
    "print(rdd.groupBy(lambda x: \"group {}\".format(x % 3)).map(lambda x: (x[0], list(x[1]))).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "1a1be49b-e498-4e01-aff8-018b26264b5d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\", 1),(\"b\", 1), (\"a\", 2), (\"b\", 2), (\"a\", 3)])\n",
    "\n",
    "print(rdd.groupBy(lambda x: x[0]).collect())\n",
    "print(rdd.groupBy(lambda x: x[0]).map(lambda x: (x[0], list(x[1]))).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "2e8d4304-6718-4934-8739-b29dc7648f01",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### groupByKey 算子\n",
    "\n",
    "groupByKey算子，**针对K-V型RDD**，自动按照K分组。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.groupByKey()\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "ca84a9ce-8380-425a-9a0a-e8eecfdee491",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\", 1),(\"b\", 1), (\"a\", 2), (\"b\", 2), (\"a\", 3)])\n",
    "\n",
    "print(rdd.groupByKey().collect())\n",
    "print(rdd.groupByKey().map(lambda x: (x[0], list(x[1]))).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "3e279d5a-96df-441c-b2c7-de50c5dc3a37",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### filter 算子\n",
    "\n",
    "filter算子，筛选满足条件的数据。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.filter(func)\n",
    "\n",
    "# func 函数\n",
    "# 函数要求传入一个参数，返回布尔类型的返回值\n",
    "```\n",
    "\n",
    "> 当函数返回True，则记录被保留，当函数返回False，则记录被丢弃"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "f86749bd-ef37-42aa-9588-ac01063f638a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([0,1,2,3,4,5,6,7,8,9])\n",
    "print(rdd.filter(lambda x: x > 5).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "5afc4f1c-b696-49a7-a115-805a9949fcb0",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### distinct 算子\n",
    "\n",
    "distinct算子，对RDD的数据进行去重。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.distinct(参数)\n",
    "\n",
    "# 参数：去重的分区数，一般不用传\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "f5b9e83b-2202-4410-bda1-618090d333a6",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([0,1,2,0,1,2,0,2,2,0], 5)\n",
    "print(rdd.glom().collect())\n",
    "print(rdd.distinct().collect())\n",
    "print(rdd.distinct(5).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "84dff800-633a-42be-a8dc-cf72ed56b19d",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### union 算子\n",
    "\n",
    "union算子，将两个RDD合并成1个RDD。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.union(other)\n",
    "\n",
    "# other，另一个需要被合并的RDD\n",
    "```\n",
    "\n",
    "> union 只会合并，不会去重"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "dc83c7f9-5747-44e7-aaa8-4359f89f3469",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])\n",
    "rdd2 = sc.parallelize([10, 11, 12, 13, 14, 15, 16, 17, 18, 19])\n",
    "print(rdd.union(rdd2).collect())\n",
    "print(rdd.union(rdd.map(lambda x: (x, x))).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "e339c71e-79d8-4381-a5db-a4fa0b16416e",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### join 算子\n",
    "\n",
    "join算子，对两个RDD执行JOIN操作（可实现SQL的内、外连接）。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.join(other)\n",
    "rdd.leftOuterJoin(other)\n",
    "rdd.rightOuterJoin(other)\n",
    "```\n",
    "\n",
    "> join算子只能用于二元元组"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "5b391097-bb0f-4017-9955-7428936c012a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd1 = sc.parallelize([(101,\"A\"), (102, \"B\"), (103, \"C\")])\n",
    "rdd2 = sc.parallelize([(102, 90), (104, 95)])\n",
    "\n",
    "# join是内连接，保留两个RDD都有的数据\n",
    "print(rdd1.join(rdd2).collect())\n",
    "\n",
    "# leftOuterJoin是左外连接\n",
    "print(rdd1.leftOuterJoin(rdd2).collect())\n",
    "\n",
    "# rightOuterJoin是右外连接\n",
    "print(rdd1.rightOuterJoin(rdd2).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "b86f90ee-c26e-4629-b0f2-4cc5974d9bda",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### intersection 算子\n",
    "\n",
    "intersection算子，求两个RDD的交集，即同时存在的元素。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.intersection(other)\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "289084a9-bd6a-4ccd-ac76-72a95380b210",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd1 = sc.parallelize([(101, \"A\"), (102, \"B\"), (103, \"C\"), \"A\", 2])\n",
    "rdd2 = sc.parallelize([(102, 90), (104, 95), (103, \"C\"), \"A\", 1, 2])\n",
    "\n",
    "print(rdd1.intersection(rdd2).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "226acf3e-53e1-4518-881f-083b70c8a35d",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### sortBy 算子\n",
    "\n",
    "sortBy算子，对RDD的数据进行排序，基于给定的排序依据。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.sortBy(func, ascending=True, numPartitions=None)\n",
    "\n",
    "# func 函数\n",
    "# 函数要求传入一个参数，返回一个返回值\n",
    "# sortBy根据函数的返回值，将RDD的元素进行排序\n",
    "# ascending：True升序，Flase降序\n",
    "# numPartitions：排序后的分区数\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "246c523d-f135-49b0-8670-a5b248ba5684",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([0,1,3,7,4,5,8,2,6,9])\n",
    "\n",
    "print(rdd.sortBy(lambda x: x).glom().collect())\n",
    "print(rdd.sortBy(lambda x: x, False).glom().collect())\n",
    "print(rdd.sortBy(lambda x: x % 3, False).glom().collect())\n",
    "print(rdd.sortBy(lambda x: x % 3, False, 3).glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "e4333eec-937b-4a21-94ff-8d5b08e7796a",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### sortByKey 算子\n",
    "\n",
    "sortByKey算子，**针对K-V型RDD**，按照K对RDD的数据进行排序。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.sortByKey(ascending=True, numPartitions=None, func)\n",
    "\n",
    "# ascending：True升序，Flase降序\n",
    "# numPartitions：排序的分区数\n",
    "# func 函数\n",
    "# 函数要求传入一个参数，返回一个返回值\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "520a134b-f3a1-4502-83a1-11959215a859",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\", 1), (\"b\", 1), (\"a\", 3), (\"c\", 4), (\"b\", 2), (\"a\", 2)])\n",
    "\n",
    "print(rdd.sortByKey().glom().collect())\n",
    "print(rdd.sortByKey(False).glom().collect())\n",
    "print(rdd.sortByKey(False, 2).glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "5ca92066-a92c-4499-ad89-54fb049e4502",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### zip 算子\n",
    "\n",
    "zip算子，将两个RDD的元素一一对应地组成一个二元组的RDD，就像拉链一样。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.zip(other)\n",
    "\n",
    "# other：需要做拉链的另一个RDD\n",
    "```\n",
    "\n",
    "> Can only zip with RDD which has the same number of partitions  \n",
    "Can only zip RDDs with same number of elements in each partition  \n",
    "需要确保做zip的两个RDD的元素个数相同。  \n",
    "即便是用Scala编写的Spark代码也要注意，确保两个RDD的元素个数相同，这一点与Scala的List的zip不同。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "f15a4b34-491b-4317-958f-d416890e2b37",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([1,2,3,4])\n",
    "\n",
    "# Can only zip with RDD which has the same number of partitions\n",
    "# print(rdd.zip(sc.parallelize([\"a\",\"b\",\"c\",\"d\"],3)).collect())\n",
    "\n",
    "# Can only zip RDDs with same number of elements in each partition\n",
    "# print(rdd.zip(sc.parallelize([\"a\",\"b\",\"c\"])).collect())\n",
    "# print(rdd.zip(sc.parallelize([\"a\",\"b\",\"c\",\"d\",\"e\"])).collect())\n",
    "\n",
    "print(rdd.zip(sc.parallelize([\"a\",\"b\",\"c\",\"d\"])).collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "790e58db-db84-45c7-a5ac-a490e03689a4",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%scala\n",
    "val list1 = List(1,2,3,4)\n",
    "val list2 = List(\"a\",\"b\",\"c\")\n",
    "val list3 = List(\"a\",\"b\",\"c\",\"d\",\"e\")\n",
    "println(list1.zip(list2))\n",
    "println(list1.zip(list3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "a01a2e2b-7295-45e1-8540-2babcdc57e75",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "%scala\n",
    "\n",
    "val rdd1 = sc.parallelize(List(1,2,3,4))\n",
    "val rdd2 = sc.parallelize(List(\"a\",\"b\",\"c\",\"d\",\"e\"))\n",
    "\n",
    "println(rdd1.zip(rdd2).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "c6da9a85-83ec-474e-9731-b57bfcc5e2d4",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### zipWithIndex 算子\n",
    "\n",
    "zipWithIndex算子，将RDD的元素与每个元素对应的索引进行zip。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.zipWithIndex()\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "13fa2377-4009-41f2-95c2-ffc83979849c",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([\"a\",\"b\",\"c\",\"d\"])\n",
    "\n",
    "print(rdd.zipWithIndex().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "928af30e-51f9-491e-b32f-b60ebde9a796",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### zipWithUniqueId 算子\n",
    "\n",
    "zipWithUniqueId算子，将RDD的元素与不重复的ID进行zip。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.zipWithUniqueId()\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "7a8ee124-2231-4663-bccc-f02f02de203f",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([\"a\",\"b\",\"c\",\"d\",\"e\",\"f\"])\n",
    "\n",
    "print(rdd.zipWithUniqueId().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "ddb55066-ccf0-4647-9483-45d596dadab7",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### partitionBy 算子\n",
    "\n",
    "partitionBy算子，**针对K-V型RDD**，根据K对RDD进行自定义分区操作。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.partitionBy(numPartitions, partitionFunc)\n",
    "\n",
    "# numPartitions：重新分区后的分区数\n",
    "# partitionFunc：自定义分区规则，返回的是整数类型的分区编号，取值范围: [0, numPartitions - 1]，返回值不在这个取值范围的，默认都放最后一个分区\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "187bc21a-f82f-46b2-836e-45170487432d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([\"a\",\"b\",\"c\",\"d\",\"e\",\"f\"]).zipWithIndex()\n",
    "\n",
    "print(rdd.glom().collect())\n",
    "\n",
    "def partition_func(key):\n",
    "    if key <= \"b\": return 1\n",
    "    if key <= \"d\": return 2\n",
    "    if key >= \"f\": return 5\n",
    "    return 0\n",
    "\n",
    "print(rdd.partitionBy(3,partition_func).glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "b91d5a8c-8805-4f2f-9f1c-be1760503d2e",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### repartition 算子\n",
    "\n",
    "repartition算子，对RDD的分区执行重新分区。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.repartition(numPartitions)\n",
    "\n",
    "# numPartitions：重新分区后的分区数\n",
    "```\n",
    "\n",
    "> 重分区采用的是：合并小分区、增加空分区的方式进行操作的。\n",
    "\n",
    "> **注意：**  对分区的数量进行操作一定要慎重。  \n",
    "> 一般情况下，除了要求全局排序设置为1个分区外，多数时候我们都不需要重新分区。  \n",
    "> 如果改变了分区，可能会影响**并行计算**，还可能导致shuffle。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "2b81c23a-9460-400b-810b-d0bc7944f78a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([\"a\",\"b\",\"c\",\"d\",\"e\",\"f\"])\n",
    "\n",
    "print(rdd.glom().collect())\n",
    "print(rdd.repartition(2).glom().collect())\n",
    "print(rdd.repartition(9).glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "c075ec04-fbff-4186-84a2-485bd92faec3",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### coalesce 算子\n",
    "\n",
    "coalesce算子，对分区数量进行增减。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.coalesce(numPartitions, shuffle=False)\n",
    "\n",
    "# numPartitions：重新分区后的分区数\n",
    "# shuffle：表示是否允许shuffle，增加分区数会导致shuffle\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "b65a7f20-661f-4718-b04a-fe624b1d79e9",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([\"a\",\"b\",\"c\",\"d\",\"e\",\"f\"])\n",
    "\n",
    "print(rdd.glom().collect())\n",
    "print(rdd.coalesce(2).glom().collect())\n",
    "print(rdd.coalesce(6).glom().collect())\n",
    "print(rdd.coalesce(6, True).glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "befa1ba0-1f50-4ec8-ac87-e06f494f98f1",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### mapPartitions 算子\n",
    "\n",
    "mapPartitions算子，和map一致，对RDD执行指定的逻辑操作，一次处理一整个分区。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.mapPartitions(func, preservesPartitioning=False)\n",
    "\n",
    "# func：处理函数，由于每次处理的是一整个分区的数据，处理的是迭代数据，返回值也要求是可迭代的数据\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "67cc6d15-3c6c-4766-a239-595dafbb504f",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([1,2,3,4,5,6,7,8,9])\n",
    "\n",
    "print(rdd.glom().collect())\n",
    "print(rdd.map(lambda x: (type(x), x)).glom().collect())\n",
    "print(rdd.mapPartitions(lambda x: [type(x)]).glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "39c158dd-d550-4e84-b1fb-fb19abcf58ed",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### 案例\n",
    "\n",
    "从订单列表中统计 Plain Papadum 每年的销售额。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "45968a94-1eaf-4497-bd00-ceffdaca0550",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "import datetime\n",
    "\n",
    "# 读取数据文件\n",
    "fileRdd = sc.textFile(\"/mnt/databrickscontainer1/restaurant-1-orders.csv\")\n",
    "\n",
    "# 将订单中的标题行去掉\n",
    "dataRdd = fileRdd.map(lambda x: x.split(\",\")).filter(lambda x: x[0] != \"Order Number\")\n",
    "\n",
    "# 过滤出 Plain Papadum 的订单\n",
    "papadumRdd = dataRdd.filter(lambda x: x[2] == \"Plain Papadum\")\n",
    "\n",
    "# 将日期字符串转换成日期，将数量与单价相乘\n",
    "dateRdd = papadumRdd.map(lambda x: [datetime.datetime.strptime(x[1],'%d/%m/%Y %H:%M').year, float(x[3]) * float(x[4])])\n",
    "\n",
    "# 将订单金额按年进行聚合\n",
    "totalRdd = dateRdd.reduceByKey(lambda a,b: a + b)\n",
    "\n",
    "# 原始统计结果\n",
    "print(totalRdd.collect())\n",
    "# 统计结果按年排序\n",
    "print(totalRdd.sortByKey().collect())\n",
    "# 统计结果按销售额排序\n",
    "print(totalRdd.sortBy(lambda x: x[1]).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "aa94f111-eb3b-4f56-aa51-b2b34d4bc34e",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### 常用的Action算子"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "39545c25-7611-47fb-9194-e82d8f8947ac",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### collect 算子\n",
    "\n",
    "collect算子，将RDD各个分区内的数据，统一收集到Driver中，形成List对象。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.collect()\n",
    "```\n",
    "\n",
    "> 这个算子，是将RDD各个分区的数据都拉取到Driver中  \n",
    "RDD是分布式对象，其数据可能很大，所以用这个算子之前，需要了解RDD的数据，确保数据集不会太大，否则会吧Driver的内存撑爆。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "1ef2c2b2-fbdc-41bd-af88-df6ab407bbc8",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([0,1,2,0,1,2,0,2,2,0], 2)\n",
    "print(rdd)\n",
    "print(rdd.collect())\n",
    "print(rdd.glom().collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "b34dc1c0-ee09-44ed-bcfb-012fa9494ff8",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### countByKey 算子\n",
    "\n",
    "countByKey算子，**针对K-V型RDD**，统计K出现的次数。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.countByKey()\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "fc20bd3d-b710-4f7d-adb7-e3476ffea8ca",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize([(\"a\", 1), (\"b\", 1), (\"a\", 3), (\"c\", 4), (\"b\", 2), (\"a\", 2)])\n",
    "\n",
    "print(rdd.countByKey())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "17b86982-36b1-4af3-aeeb-bd2edf401523",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### reduce 算子\n",
    "\n",
    "reduce算子，对RDD数据集按照指定的逻辑进行聚合。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.reduce(func)\n",
    "\n",
    "# func：(T, T) -> T\n",
    "# 接收两个参数，返回一个返回值，要求参数与返回值的类型相同\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "5f323675-c95c-46b1-b5c9-254ca2dbc93c",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize(range(1, 11), 1)\n",
    "\n",
    "print(rdd.reduce(lambda a, b: a + b))\n",
    "print(rdd.reduce(lambda a, b: a * b))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "f1699533-7364-4b3f-9b11-2343a377c055",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### fold 算子\n",
    "\n",
    "fold算子，使用一个初始值，对RDD数据集按照指定的逻辑进行聚合。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.fold(zeroValue, func)\n",
    "\n",
    "# zeroValue：T，初始值\n",
    "# func：(T, T) -> T\n",
    "# 初始值、函数的参数、函数的返回值，类型相同\n",
    "```\n",
    "\n",
    "> **注意：** fold的初始值，会同时作用在  \n",
    "> * 分区内聚合\n",
    "> * 分区间聚合\n",
    "> * 即便只有1个分区，也会将初始值作用在分区上"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "ec9e0230-35b6-41fa-8a6b-ae5d04177c58",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# 当有1个分区时：初始值 + 分区1的聚合结果（初始值 + 分区元素的聚合结果） = 5 + (5 + 5050) = 5060\n",
    "print(sc.parallelize(range(1,101),1).fold(5, lambda a,b: a + b))\n",
    "# 当有2个分区时：(x1 + x2 = 5050)\n",
    "# 初始值 + 分区1的聚合结果（初始值 + 分区元素的聚合结果） + 分区2的聚合结果（初始值 + 分区元素的聚合结果） = 5 + (5 + x1) + (5 + x2) = 5065\n",
    "print(sc.parallelize(range(1,101),2).fold(5, lambda a,b: a + b))\n",
    "# 当有3个分区时：(x1 + x2 + x3 = 5050)\n",
    "# 初始值 + 分区1的聚合结果（初始值 + 分区元素的聚合结果） + 分区2的聚合结果（初始值 + 分区元素的聚合结果） + 分区3的聚合结果（初始值 + 分区元素的聚合结果） = 5 + (5 + x1) + (5 + x2) + (5 + x3) = 5070\n",
    "print(sc.parallelize(range(1,101),3).fold(5, lambda a,b: a + b))\n",
    "# 由于我们的CPU是4个，所以默认是4个分区，结果就是 5075\n",
    "print(sc.parallelize(range(1,101),4).fold(5, lambda a,b: a + b))\n",
    "# 分区数每增加1个，最终聚合的结果就增加一个初始值\n",
    "print(sc.parallelize(range(1,101)).fold(5, lambda a,b: a + b))\n",
    "print(sc.parallelize(range(1,101)).glom().collect())\n",
    "print(sc.parallelize(range(1,101)).getNumPartitions())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "38f788f0-fb71-4949-b9b6-f5d72673924f",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "print(sc.parallelize([\"a\",\"b\",\"c\"],1).fold(\"f\", lambda a,b: a + b))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\"],2).fold(\"f\", lambda a,b: a + b))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\"],3).fold(\"f\", lambda a,b: a + b))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\"]).fold(\"f\", lambda a,b: a + b))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\"],50).fold(\"f\", lambda a,b: a + b))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "05616127-4a8a-43c9-8928-8417b13d3dc8",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### first 算子\n",
    "\n",
    "first算子，取出RDD中的第一个元素。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.first()\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "2445550d-1d2f-4a54-847f-2a66f43a3c6a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "print(sc.parallelize([\"a\",\"b\",\"c\"],1).first())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "25cf2d2e-ce23-45be-b75e-a4da15aeea4c",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### take 算子\n",
    "\n",
    "take算子，取出RDD中的前N个元素。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.take(N)\n",
    "\n",
    "# N：需要取出的元素个数\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "46367528-4d41-4b1b-a4de-f712c6c90a47",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).take(3))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],2).glom().collect(), sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],2).take(3))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],3).glom().collect(), sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],3).take(3))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],4).glom().collect(), sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],4).take(3))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],5).glom().collect(), sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],5).take(3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "50f171d4-17aa-4593-a4b1-99a290adc30b",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### top 算子\n",
    "\n",
    "top算子，对RDD中的元素进行降序排序，然后取出前N个元素。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.top(N)\n",
    "\n",
    "# N：需要取出的元素个数\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "1c803341-95eb-4230-bb14-a55314a9185a",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).top(3))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],2).glom().collect(), sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],2).top(3))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],3).glom().collect(), sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],3).top(3))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],4).glom().collect(), sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],4).top(3))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],5).glom().collect(), sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],5).top(3))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "932b63a4-3adb-4ecf-9e53-1ac97caed827",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### takeSample 算子\n",
    "\n",
    "takeSample算子，随机抽样RDD的数据。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.takeSample(withReplacement, num, seed=None)\n",
    "\n",
    "# withReplacement：是否允许数据重复，True表示可以多次取同一个数据，False表示不可以取同一个数据\n",
    "# num：抽样数目\n",
    "# seed：随机数种子\n",
    "```\n",
    "\n",
    "> 随机数种子数字可以随便传，相同随机数种子取出的结果是一致的。一般我们不给定这个参数，由Spark自己给定。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "32f3b51a-8e3d-4f2a-afdd-da777b5ba9ee",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# 允许多次抽取同一个元素\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(True,5))\n",
    "# 不允许多次抽取同一个元素，但每次抽取的结果可能是不一样的\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5))\n",
    "# 指定随机数种子后，每次抽取的结果是一致的\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5, 5))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5, 5))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5, 5))\n",
    "# 不同的随机数种子，抽取的结果是不一样的\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5, 5))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5, 4))\n",
    "print(sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1).takeSample(False,5, 2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "2b1756e4-9cec-4e86-9b2e-c93ece29d216",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### takeOrdered 算子\n",
    "\n",
    "takeOrdered算子，对RDD进行排序，然后取出前N个元素，与top类似，只是可以自己指定排序规则。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.takeOrdered(num, func)\n",
    "\n",
    "# num：需要取出的元素个数\n",
    "# func：排序规则\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "6dab3283-c488-48f6-8e92-22f71672106d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "print(sc.parallelize([\"aaa\", \"eee\", \"ff\", \"bbbbb\", \"g\", \"cccc\"], 1).top(3))\n",
    "print(sc.parallelize([\"aaa\", \"eee\", \"ff\", \"bbbbb\", \"g\", \"cccc\"], 1).takeOrdered(3, lambda x: -len(x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "eb2f02a2-f3ee-4462-896c-c2b4fb96c3b8",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### foreach 算子\n",
    "\n",
    "foreach算子，对RDD每一个元素，执行指定的逻辑操作，这个算子没有返回值。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.foreach(func)\n",
    "\n",
    "# func: (T) -> None\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "7c8e85b7-cf94-4a8f-941b-4118144f1dfd",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "print(sc.parallelize(range(1,11),1).map(lambda x: x * 2))\n",
    "print(sc.parallelize(range(1,11),1).map(lambda x: x * 2).collect())\n",
    "print(sc.parallelize(range(1,11),1).foreach(lambda x: print(x, '-->', x * 2)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "2fb284d3-9e96-48bd-84cb-2b3c23ccd957",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### foreachPartition 算子\n",
    "\n",
    "foreachPartition算子，和foreach一致，对RDD每一个元素，执行指定的逻辑操作，一次处理一整个分区，这个算子没有返回值。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.foreachPartition(func)\n",
    "\n",
    "# func: (T) -> None\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "6b13d63b-f2c7-46a5-adce-7b7ee1bf4eeb",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd = sc.parallelize(range(1,11),3)\n",
    "\n",
    "rdd.foreach(lambda x: print(x, type(x)))\n",
    "rdd.foreachPartition(lambda x: print(type(x), list(x)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "ffdd92f0-fe86-4add-a040-52f90b6ca003",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### saveAsTextFile 算子\n",
    "\n",
    "saveAsTextFile算子，将RDD的数据写入文本文件中。支持写到本地、分布式文件系统等，每个分区写一个子文件。\n",
    "\n",
    "语法：\n",
    "```\n",
    "rdd.saveAsTextFile(path,compressionCodecClass=None)\n",
    "\n",
    "# path: 文件的写出路径\n",
    "# compressionCodecClass：压缩类\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "09718c4a-af96-43e6-9636-2fbe58a5c728",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "rdd1 = sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],1)\n",
    "print(rdd1.glom().collect())\n",
    "rdd1.saveAsTextFile(\"/mnt/databrickscontainer1/partition1\")\n",
    "\n",
    "rdd2 = sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],2)\n",
    "print(rdd2.glom().collect())\n",
    "rdd2.saveAsTextFile(\"/mnt/databrickscontainer1/partition2\")\n",
    "\n",
    "rdd3 = sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],3)\n",
    "print(rdd3.glom().collect())\n",
    "rdd3.saveAsTextFile(\"/mnt/databrickscontainer1/partition3\")\n",
    "\n",
    "rdd4 = sc.parallelize([\"a\",\"b\",\"c\",\"e\",\"f\",\"g\"],4)\n",
    "print(rdd4.glom().collect())\n",
    "rdd4.saveAsTextFile(\"/mnt/databrickscontainer1/partition4\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "inputWidgets": {},
     "nuid": "8607fea8-7743-4ac9-8779-9e836a6e369b",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "#### 注意\n",
    "\n",
    "在前面的Action算子中，有几个算子：\n",
    "* foreach\n",
    "* foreachPartition\n",
    "* saveAsTextFile\n",
    "\n",
    "这几个算子无返回值，是分区直接执行的，跳过Driver，由Executor直接执行。\n",
    "\n",
    "其他算子有返回值，都会将执行结果发送到Driver。"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "RDD编程入门",
   "notebookOrigID": 2180139033493940,
   "widgets": {}
  },
  "kernelspec": {
   "display_name": "",
   "name": ""
  },
  "language_info": {
   "name": ""
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
